package cn.karent.processor;

import cn.karent.common.DatabaseConfig;
import cn.karent.common.JsonUtils;
import cn.karent.common.ValidationUtils;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
import com.fasterxml.jackson.databind.annotation.JsonNaming;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotEmpty;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.helpers.MessageFormatter;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import static cn.karent.processor.MysqlSinkProcessor.BEAN_NAME;

/**
 * 数据写入到mysql
 *
 * @author wanshengdao
 * @date 2025/3/20
 */
@Slf4j
@Component(value = BEAN_NAME)
@RequiredArgsConstructor
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class MysqlSinkProcessor extends ConfigurableProcessor<MysqlSinkProcessor.Config> {

    public static final String BEAN_NAME = "MysqlSink";

    private final DatabaseConfig databaseConfig;

    private static volatile HikariDataSource dataSource;

    /**
     * 获取数据源
     *
     * @return 数据源
     */
    private DataSource getDataSource() {
        if (dataSource == null) {
            synchronized (MysqlSinkProcessor.class) {
                if (dataSource == null) {
                    ValidationUtils.valid(databaseConfig);
                    HikariConfig config = new HikariConfig();
                    config.setJdbcUrl(databaseConfig.getUrl());
                    config.setUsername(databaseConfig.getUsername());
                    config.setPassword(databaseConfig.getPassword());
                    config.setMaximumPoolSize(10);
                    dataSource = new HikariDataSource(config);
                }
            }
        }
        return dataSource;
    }


    @Override
    public void process(String msg) {
        Map<String, Object> parameter = JsonUtils.parseMap(msg);
        String sql = generateSql();
        log.info("sql: {}", sql);
        // 执行插入
        execute(sql, parameter);
    }

    private void execute(String sql, Map<String, Object> parameter) {
        Connection connection = null;
        PreparedStatement ps = null;
        try {
            connection = getDataSource().getConnection();
            ps = connection.prepareStatement(sql);
            List<String> columns = c.getColumns();
            for (int i = 0; i < columns.size(); i++) {
                String column = columns.get(i);
                setParameter(ps, i + 1, parameter.get(column));
            }
            int affectRows = ps.executeUpdate();
            log.info("插入成功, 影响行数: {}", affectRows);
        } catch (Exception e) {
            log.error("异常: ", e);
        } finally {
            try {
                if (ps != null) {
                    ps.close();
                }
                if (connection != null) {
                    connection.close();
                }
            } catch (SQLException e) {
                log.error("异常: ", e);
            }
        }
    }

    private static void setParameter(PreparedStatement ps, int idx, Object o) throws SQLException {
        if (Objects.isNull(o)) {
            ps.setNull(idx, Types.NULL);
            return;
        }
        if (o instanceof Integer tmp) {
            ps.setInt(idx, tmp);
        } else if (o instanceof String tmp) {
            ps.setString(idx, tmp);
        } else {
            throw new IllegalStateException(String.format("未知的类型 %s", o.getClass()));
        }
    }

    private String generateSql() {
        StringBuilder sb = new StringBuilder();
        sb.append("insert {}(");
        List<String> columns = c.getColumns();
        sb.append("{},".repeat(columns.size()));
        sb.deleteCharAt(sb.length() - 1);
        sb.append(") values(");
        sb.append("?,".repeat(columns.size()));
        sb.deleteCharAt(sb.length() - 1);
        sb.append(")");
        List<Object> param = new ArrayList<>();
        param.add(c.getTableName());
        param.addAll(columns);
        return MessageFormatter.arrayFormat(sb.toString(), param.toArray(new Object[0])).getMessage();
    }

    @Setter
    @Getter
    @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
    public static class Config {

        @NotBlank(message = "表名不能为空")
        private String tableName;

        @NotEmpty(message = "列名不能为空")
        private List<String> columns;

    }

}
